package csl.mq.rabbitmq;

import com.rabbitmq.client.*;
import csl.mq.rabbitmq.utils.RabbitMQUtils;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Description: TODO 消费者
 * @author: 隔山海
 * @date: 2021年08月30日 21:54
 */
@SpringBootTest
@Slf4j
public class Consumer {

    /**
     * 测试直连式消费者
     */
    @Test
    public void receiveMessage() throws IOException, TimeoutException {
//        创建连接工厂并初始化连接
        Connection connection = RabbitMQUtils.initRabbitMQ();
//        创建通道
        Channel channel = connection.createChannel();
        /*
         * 绑定队列
         * 参数1：队列名
         * 参数2：是否持久化
         * 参数3：是否独享
         * 参数4：是否自动删除
         * 参数5：其他参数
         * */
        channel.queueDeclare("test", true, false, false, null);
        /*
         * 接收消息
         * 参数1：队列名
         * 参数2：是否自动签收
         * 参数3：回调接口
         * */
        channel.basicConsume("test", true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("消费消息===>{}", new String(body));
            }
        });
        System.in.read();
    }

    /**
     * Work Queue模型消费者
     */
    @Test
    public void receiveWorkQueueConsumer1() throws IOException, TimeoutException {
        Connection connection = RabbitMQUtils.initRabbitMQ();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work-queue", false, false, false, null);
        channel.basicConsume("work-queue", true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                log.info("workQueue[1]消费消息==={}", new String(body));
            }
        });
        System.out.println("workQueue[1]启动！");
        System.in.read();
    }

    /**
     * Work Queue模型消费者2
     */
    @Test
    public void receiveWorkQueueConsumer2() throws IOException, TimeoutException {
        Connection connection = RabbitMQUtils.initRabbitMQ();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work-queue", false, false, false, null);
//        一次只处理一条消息
        channel.basicQos(100);
        /*
        * 参数1：队列名称
        * 参数2：关闭自动确认消息
        * */
        channel.basicConsume("work-queue",false,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("workQueue[2]消费消息==={}", new String(body));
                /*手动确认消息*/
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        });
        System.out.println("workQueue[2]启动！");
        System.in.read();
    }


    /**
     * Fanout任务模型消费者1
     */
    @Test
    public void receiveFanout1() throws IOException, TimeoutException {
        Connection connection = RabbitMQUtils.initRabbitMQ();
        Channel channel = connection.createChannel();
        /*绑定交换机*/
        channel.exchangeDeclare("logs",BuiltinExchangeType.FANOUT);
        /*从交换机获取一个临时队列*/
        String queue = channel.queueDeclare().getQueue();
        /*把临时队列和交换机绑定*/
        channel.queueBind(queue,"logs","");
        /*接收消息*/
        channel.basicConsume(queue,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("消费消息【1】====>{}",new String(body));
            }
        });
        System.err.println("Fanout消费者【1】启动成功！");
        System.in.read();
    }

    /**
     * Fanout任务模型消费者2
     */
    @Test
    public void receiveFanout2() throws IOException, TimeoutException {
        Connection connection = RabbitMQUtils.initRabbitMQ();
        Channel channel = connection.createChannel();
        /*绑定交换机*/
        channel.exchangeDeclare("logs",BuiltinExchangeType.FANOUT);
        /*从交换机获取一个临时队列*/
        String queue = channel.queueDeclare().getQueue();
        /*把临时队列和交换机绑定*/
        channel.queueBind(queue,"logs","");
        /*接收消息*/
        channel.basicConsume(queue,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("消费消息【2】====>{}",new String(body));
            }
        });
        System.err.println("Fanout消费者【2】启动成功！");
        System.in.read();
    }


    /**
     * Routing direct:测试Direct消费端1
     */
    @Test
    public void receiveRoutingDirect1() throws IOException, TimeoutException {
        Connection connection = RabbitMQUtils.initRabbitMQ();
        Channel channel = connection.createChannel();
        String exChangeName = "routing-key";
        channel.exchangeDeclare(exChangeName,BuiltinExchangeType.DIRECT);
        /*获取一个临时队列*/
        String queue = channel.queueDeclare().getQueue();
        /*绑定队列到交换机*/
        channel.queueBind(queue,exChangeName,"error");
        /*消费消息*/
        channel.basicConsume(queue,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.err.println("error消费者消费消息===>"+new String(body));
            }
        });
        System.in.read();
    }

    /**
     * Routing direct:测试Direct消费端2
     */
    @Test
    public void receiveRoutingDirect2() throws IOException, TimeoutException {
        Connection connection = RabbitMQUtils.initRabbitMQ();
        Channel channel = connection.createChannel();
        String exChangeName = "routing-key";
        channel.exchangeDeclare(exChangeName,BuiltinExchangeType.DIRECT);
        /*获取一个临时队列*/
        String queue = channel.queueDeclare().getQueue();
        /*绑定队列到交换机*/
        channel.queueBind(queue,exChangeName,"error");
        channel.queueBind(queue,exChangeName,"info");
        channel.queueBind(queue,exChangeName,"warn");
        /*消费消息*/
        channel.basicConsume(queue,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                System.err.println("渣男消费者消费消息===>"+new String(body));
            }
        });
        System.in.read();
    }


    /**
     * RoutingTopic1:测试直Topic消费端1
     */
    @Test
    public void receiveRoutingTopic1() throws IOException, TimeoutException {
        Connection connection = RabbitMQUtils.initRabbitMQ();
        Channel channel = connection.createChannel();
        String exChangeName = "topic";
        channel.exchangeDeclare(exChangeName,BuiltinExchangeType.TOPIC);
        /*获取一个临时队列*/
        String queue = channel.queueDeclare().getQueue();
        /*绑定队列到交换机*/
        channel.queueBind(queue,exChangeName,"user.*");
        /*消费消息*/
        channel.basicConsume(queue,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                System.err.println("渣男消费者消费消息===>"+new String(body));
            }
        });
        System.in.read();
    }

    /**
     * RoutingTopic1:测试Topic消费端2
     */
    @Test
    public void receiveRoutingTopic2() throws IOException, TimeoutException {
        Connection connection = RabbitMQUtils.initRabbitMQ();
        Channel channel = connection.createChannel();
        String exChangeName = "topic";
        channel.exchangeDeclare(exChangeName,BuiltinExchangeType.TOPIC);
        /*获取一个临时队列*/
        String queue = channel.queueDeclare().getQueue();
        /*绑定队列到交换机*/
        channel.queueBind(queue,exChangeName,"user.#");
        /*消费消息*/
        channel.basicConsume(queue,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                System.err.println("渣男消费者消费消息===>"+new String(body));
            }
        });
        System.in.read();
    }

    /**
     *
     * @throws IOException
     * @throws TimeoutException
     */
    @Test
    public void receiveConfirm() throws IOException, TimeoutException {
        Connection connection = RabbitMQUtils.initRabbitMQ();
        Channel channel = connection.createChannel();
        String exChangeName = "confirm";
        channel.exchangeDeclare(exChangeName,BuiltinExchangeType.DIRECT);
        /*获取一个临时队列*/
        String queue = channel.queueDeclare().getQueue();
        /*绑定队列到交换机*/
        channel.queueBind(queue,exChangeName,"confirmError");
        /*消费消息*/
        channel.basicConsume(queue,new DefaultConsumer(channel){

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                System.err.println("渣男消费者消费消息===>"+new String(body));
            }
        });
        System.in.read();
    }
}
